package com.aiwiown.snackmq.storage.file;

import com.aiwiown.snackmq.storage.config.StorageConfig;
import com.aiwiown.snackmq.storage.util.BufferCleaner;
import com.aiwiown.snackmq.storage.util.ResourceManager;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class MappedFile {

    // 文件名
    private final String fileName;
    // 文件对象
    private final File file;
    // 文件通道
    private FileChannel fileChannel;
    // 内存映射字节缓冲区
    private MappedByteBuffer mappedByteBuffer;
    // 写入位置
    private final AtomicLong wrotePosition;
    // 提交位置
    private final AtomicLong committedPosition;
    // 刷盘位置
    private final AtomicLong flushedPosition;
    // 存储配置
    private final StorageConfig config;
    // 随机访问文件
    private RandomAccessFile randomAccessFile;
    // 映射文件大小
    private final long mappedFileSize;
    // 是否已关闭
    private volatile boolean closed = false;
    // 是否已初始化
    private volatile boolean initialized = false;

    // 使用ConcurrentHashMap的keySet实现线程安全的Set
    private final Set<Long> invalidOffsets = ConcurrentHashMap.newKeySet();

    // 读锁
    private final Lock readLock = new ReentrantLock();
    // 写锁
    private final Lock writeLock = new ReentrantLock();

    // 内存池相关字段
    private static final int BUFFER_POOL_SIZE = 100;
    private static final int BUFFER_SIZE = 8192; // 8KB缓冲区
    private final Queue<ByteBuffer> bufferPool;
    private final ReentrantLock bufferPoolLock = new ReentrantLock();

    /**
     * 构造函数，初始化映射文件
     *
     * @param fileName 文件名
     * @param config   存储配置
     * @throws IOException 如果文件创建或初始化失败
     */
    public MappedFile(String fileName, StorageConfig config) throws IOException {
        this.fileName = fileName;
        this.config = config;
        this.mappedFileSize = config.getMappedFileSize();
        this.wrotePosition = new AtomicLong(0);
        this.committedPosition = new AtomicLong(0);
        this.flushedPosition = new AtomicLong(0);

        // 初始化内存池
        this.bufferPool = new ConcurrentLinkedQueue<>();
        initializeBufferPool();

        this.file = new File(config.getMessageStorePath(), fileName);
        if (!this.file.getParentFile().exists()) {
            this.file.getParentFile().mkdirs();
        }

        this.randomAccessFile = new RandomAccessFile(this.file, "rw");
        this.fileChannel = this.randomAccessFile.getChannel();
        this.mappedByteBuffer = this.fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, this.mappedFileSize);

        // 预热文件
        warmup();

        // 恢复状态
        recover();

        this.initialized = true;
    }

    /**
     * 初始化内存池
     */
    private void initializeBufferPool() {
        for (int i = 0; i < BUFFER_POOL_SIZE; i++) {
            bufferPool.offer(ByteBuffer.allocateDirect(BUFFER_SIZE));
        }
    }

    /**
     * 从内存池获取ByteBuffer
     */
    private ByteBuffer getBufferFromPool() {
        ByteBuffer buffer = bufferPool.poll();
        if (buffer == null) {
            buffer = ByteBuffer.allocateDirect(BUFFER_SIZE);
        }
        buffer.clear();
        return buffer;
    }

    /**
     * 将ByteBuffer归还到内存池
     */
    private void returnBufferToPool(ByteBuffer buffer) {
        if (buffer != null && buffer.capacity() == BUFFER_SIZE) {
            buffer.clear();
            bufferPool.offer(buffer);
        }
    }

    /**
     * 批量写入结果
     */
    public static class BatchWriteResult {
        private final List<MessageWriteInfo> messageInfos;
        private final boolean success;
        private final int totalWritten;

        public BatchWriteResult(List<MessageWriteInfo> messageInfos, boolean success, int totalWritten) {
            this.messageInfos = messageInfos;
            this.success = success;
            this.totalWritten = totalWritten;
        }

        public List<MessageWriteInfo> getMessageInfos() {
            return messageInfos;
        }

        public boolean isSuccess() {
            return success;
        }

        public int getTotalWritten() {
            return totalWritten;
        }
    }

    /**
     * 单条消息写入信息
     */
    public static class MessageWriteInfo {
        private final int fileIndex;
        private final long fileOffset;
        private final int messageSize;
        private final long logicalOffset;

        public MessageWriteInfo(int fileIndex, long fileOffset, int messageSize, long logicalOffset) {
            this.fileIndex = fileIndex;
            this.fileOffset = fileOffset;
            this.messageSize = messageSize;
            this.logicalOffset = logicalOffset;
        }

        public int getFileIndex() {
            return fileIndex;
        }

        public long getFileOffset() {
            return fileOffset;
        }

        public int getMessageSize() {
            return messageSize;
        }

        public long getLogicalOffset() {
            return logicalOffset;
        }
    }

    /**
     * 优化的批量写入方法 - 使用内存池减少分配开销，返回详细的写入信息
     */
    public BatchWriteResult writeBatchMessages(List<byte[]> messages) {
        if (closed || !initialized) {
            return new BatchWriteResult(new ArrayList<>(), false, 0);
        }

        writeLock.lock();
        try {
            ByteBuffer tempBuffer = getBufferFromPool();
            int totalWritten = 0;
            List<MessageWriteInfo> messageInfos = new ArrayList<>();
            long baseLogicalOffset = getCurrentLogicalOffset();

            for (int i = 0; i < messages.size(); i++) {
                byte[] message = messages.get(i);
                // 验证原始消息格式（不包含长度前缀）
                if (message == null || message.length == 0) {
                    log.warn("Skipping null or empty message at index: {}", i);
                    continue;
                }

                int messageLength = message.length;
                int currentPosition = (int) wrotePosition.get();
                long messageLogicalOffset = baseLogicalOffset + totalWritten;

                // 检查是否需要切换缓冲区
                if (tempBuffer.remaining() < messageLength + 4) {
                    // 写入当前缓冲区内容
                    ((Buffer)tempBuffer).flip();
                    mappedByteBuffer.position(currentPosition);
                    mappedByteBuffer.put(tempBuffer);
                    wrotePosition.addAndGet(tempBuffer.limit());
                    totalWritten += tempBuffer.limit();

                    // 获取新的缓冲区
                    returnBufferToPool(tempBuffer);
                    tempBuffer = getBufferFromPool();
                    currentPosition = (int) wrotePosition.get();
                }

                // 记录消息写入信息
                int fileIndex = getFileIndex();
                long fileOffset = currentPosition;
                int messageSize = messageLength + 4; // 包含长度前缀

                // 写入消息长度和内容
                tempBuffer.putInt(messageLength);
                tempBuffer.put(message);

                // 添加到消息信息列表
                messageInfos.add(new MessageWriteInfo(fileIndex, fileOffset, messageSize, messageLogicalOffset));
            }

            // 写入剩余的缓冲区内容
            if (tempBuffer.position() > 0) {
                ((Buffer)tempBuffer).flip();
                mappedByteBuffer.position((int) wrotePosition.get());
                mappedByteBuffer.put(tempBuffer);
                wrotePosition.addAndGet(tempBuffer.limit());
                totalWritten += tempBuffer.limit();
            }

            returnBufferToPool(tempBuffer);
            commit();

            boolean success = totalWritten > 0;
            return new BatchWriteResult(messageInfos, success, totalWritten);
        } catch (Exception e) {
            log.error("Error writing batch messages", e);
            return new BatchWriteResult(new ArrayList<>(), false, 0);
        } finally {
            writeLock.unlock();
        }
    }

    /**
     * 优化的批量读取方法 - 使用内存池和零拷贝
     */
    public List<ByteBuffer> readBatchMessages(long startOffset, int maxMessages) throws IOException {
        List<ByteBuffer> messages = new ArrayList<>();
        if (startOffset < 0 || maxMessages <= 0) {
            return messages;
        }

        readLock.lock();
        try {
            if (startOffset >= wrotePosition.get()) {
                return messages;
            }

            mappedByteBuffer.position((int) startOffset);
            int messageCount = 0;
            long currentOffset = startOffset;

            while (messageCount < maxMessages && currentOffset < wrotePosition.get()) {
                if (mappedByteBuffer.remaining() < 4) {
                    log.debug("Not enough bytes for length prefix at position {}", currentOffset);
                    break;
                }

                // 记录长度前缀的位置
                int lengthPrefixPosition = mappedByteBuffer.position();
                int messageLength = mappedByteBuffer.getInt();
                log.debug("Read message length: {} at position: {}", messageLength, currentOffset);

                if (messageLength <= 0) {
                    log.debug("Invalid message length: {} at position: {}", messageLength, currentOffset);
                    break;
                }

                if (mappedByteBuffer.remaining() < messageLength) {
                    log.debug("Not enough bytes for message content: need {}, have {}", messageLength, mappedByteBuffer.remaining());
                    break;
                }

                // 创建ByteBuffer切片，包含完整的消息（包括长度前缀）
                // 需要从长度前缀开始，所以重置位置到长度前缀位置
                mappedByteBuffer.position(lengthPrefixPosition);
                ByteBuffer messageBuffer = mappedByteBuffer.slice();
                messageBuffer.limit(messageLength + 4); // 包含长度前缀
                messageBuffer.position(0); // 重置位置到开始
                messages.add(messageBuffer);

                // 跳过整个消息（长度前缀 + 消息内容）
                mappedByteBuffer.position(lengthPrefixPosition + 4 + messageLength);
                currentOffset += (4 + messageLength);
                messageCount++;
            }

            return messages;
        } finally {
            readLock.unlock();
        }
    }

    /**
     * 清理无效消息（示例实现：标记消息为无效或物理删除）
     */
    private void cleanInvalidMessages() {
        if (invalidOffsets.isEmpty()) {
            return;
        }
        // 遍历无效偏移量并清理（需根据具体存储格式实现）
        invalidOffsets.forEach(offset -> {
            // 校验offset是否在int范围内（避免Long转int溢出）
            if (offset < 0 || offset > Integer.MAX_VALUE) {
                log.error("Invalid offset for int conversion: offset={}", offset);
                return;
            }
            mappedByteBuffer.position(Math.toIntExact(offset));
            mappedByteBuffer.putInt(-1); // 标记长度为-1表示无效
        });
        // 清理完成后清空集合
        invalidOffsets.clear();
    }

    /**
     * 恢复文件状态
     */
    private void recover() {
        // 启动时清理历史无效消息
        cleanInvalidMessages();
        writeLock.lock();
        try {
            // 获取文件大小
            long fileSize = file.length();
            if (fileSize > 0) {
                // 读取文件内容
                ByteBuffer buffer = ByteBuffer.allocate((int) fileSize);
                fileChannel.read(buffer, 0);
                // 【最终修复】: 将对 .flip() 的调用强制转换为 Buffer 类型，以兼容 JDK 8
                ((Buffer) buffer).flip();
                // 将数据复制到内存映射缓冲区
                ((Buffer)mappedByteBuffer).position(0);
                mappedByteBuffer.put(buffer);

                // 扫描文件找到最后一个有效消息的结束位置
                long lastValidPosition = scanForLastValidPosition(fileSize);
                if (lastValidPosition > 0) {
                    wrotePosition.set(lastValidPosition);
                    committedPosition.set(lastValidPosition);
                    flushedPosition.set(lastValidPosition);
                    log.debug("Recovered file: {}, last valid position: {}", fileName, lastValidPosition);
                } else {
                    // 如果没有找到有效消息，重置为0
                    wrotePosition.set(0);
                    committedPosition.set(0);
                    flushedPosition.set(0);
                    log.debug("Recovered file: {}, reset to beginning", fileName);
                }
            }
        } catch (Exception e) {
            log.error("Failed to recover file: {}", fileName, e);
        } finally {
            writeLock.unlock();
        }
    }

    /**
     * 扫描文件找到最后一个有效消息的结束位置
     */
    private long scanForLastValidPosition(long fileSize) {
        try {
            long position = 0;
            while (position < fileSize - 4) { // 至少需要4字节长度前缀
                // 读取消息长度
                mappedByteBuffer.position((int) position);
                int messageLength = mappedByteBuffer.getInt();

                // 验证消息长度
                if (messageLength <= 0 || messageLength > mappedFileSize - 1024) {
                    log.debug("Invalid message length at position {}: {}", position, messageLength);
                    break;
                }

                // 检查是否有足够的空间存储这个消息
                if (position + 4 + messageLength > fileSize) {
                    log.debug("Message at position {} extends beyond file boundary", position);
                    break;
                }

                // 移动到下一个消息
                position += 4 + messageLength;
            }

            return position;
        } catch (Exception e) {
            log.error("Error scanning for last valid position", e);
            return 0;
        }
    }

    /**
     * 验证消息格式
     *
     * @param message 消息字节数组
     * @return 消息格式是否有效
     */
    private boolean isValidMessage(byte[] message) {
        if (message == null || message.length < 4) {
            log.error("Invalid message format: null or too short");
            return false;
        }

        try {
            // 读取消息长度前缀（使用大端序）
            ByteBuffer buffer = ByteBuffer.wrap(message, 0, 4);
            buffer.order(ByteOrder.BIG_ENDIAN);
            int messageLength = buffer.getInt();
            // 验证消息长度
            long mappedFileSize = config.getMappedFileSize();
            // 验证消息长度是否合理
            if (messageLength <= 0 || messageLength > mappedFileSize - 1024) {
                log.error("Invalid message length : length={}, max={}", messageLength, mappedFileSize - 1024);
                return false;
            }
            // 验证总长度是否匹配
            if (messageLength + 4 != message.length) {
                log.error("Invalid message length: expected={}, actual={}, total length={}",
                        messageLength, message.length - 4, message.length);
                return false;
            }

            return true;
        } catch (Exception e) {
            log.error("Error validating message format", e);
            return false;
        }
    }

    /**
     * 检查文件是否已满
     *
     * @return 文件是否已满
     */
    public boolean isFull() {
        return wrotePosition.get() >= config.getMappedFileSize();
    }

    /**
     * 获取剩余空间
     *
     * @return 剩余空间大小
     */
    public long getRemainingSpace() {
        return config.getMappedFileSize() - wrotePosition.get();
    }

    /**
     * 提交数据
     */
    public void commit() {
        try {
            mappedByteBuffer.force();
            flushedPosition.set(wrotePosition.get());
        } catch (Exception e) {
            log.error("Failed to commit write", e);
        }
    }

    /**
     * 刷盘
     */
    public void flush() {
        writeLock.lock();
        try {
            if (committedPosition.get() > flushedPosition.get()) {
                mappedByteBuffer.force();
                flushedPosition.set(committedPosition.get());
            }
        } catch (Exception e) {
            log.error("Error flushing file: {}", fileName, e);
        } finally {
            writeLock.unlock();
        }
    }

    /**
     * 异步刷盘
     */
    public void flushAsync() {
        if (config.isAsyncFlush() && committedPosition.get() > flushedPosition.get()) {
            writeLock.lock();
            try {
                mappedByteBuffer.force();
                flushedPosition.set(committedPosition.get());
            } catch (Exception e) {
                log.error("Error async flushing file: {}", fileName, e);
            } finally {
                writeLock.unlock();
            }
        }
    }

    /**
     * 读取消息
     *
     * @param offset 消息偏移量
     * @param length 消息长度
     * @return 消息内容，如果读取失败则返回null
     */
    public byte[] readMessage(long offset, int length) throws IOException {
        if (offset < 0 || length <= 0) {
            log.error("Invalid offset or length: offset={}, length={}", offset, length);
            return null;
        }
        // 检查是否超出文件边界
        if (offset >= wrotePosition.get()) {
            log.error("Read beyond file boundary: offset={}, wrotePosition={}", offset, wrotePosition.get());
            return null;
        }
        readLock.lock();
        try {
            // 验证总长度
            if (offset + length > wrotePosition.get()) {
                log.error("Message crosses file boundary: offset={}, length={}, wrotePosition={}",
                        offset, length, wrotePosition.get());
                return null;
            }

            // 读取整个消息（包含长度前缀）
            byte[] messageWithPrefix = new byte[length];
            mappedByteBuffer.position((int) offset);
            mappedByteBuffer.get(messageWithPrefix);

            return messageWithPrefix;
        } catch (Exception e) {
            log.error("Error reading message at offset={}, length={}", offset, length, e);
            return null;
        } finally {
            readLock.unlock();
        }
    }

    /**
     * 预热文件
     */
    private void warmup() {
        try {
            // 只在文件为空时才进行预热
            if (file.length() == 0) {
                // 写入一些数据以预热文件
                for (int i = 0; i < config.getMappedFileSize(); i += 1024) {
                    mappedByteBuffer.putInt(i, 0);
                }
                // 强制刷新到磁盘
                mappedByteBuffer.force();
                log.debug("Warmed up file: {}", fileName);
            }
        } catch (Exception e) {
            log.error("Failed to warmup file: {}", fileName, e);
        }
    }

    /**
     * 关闭映射文件
     */
    public void close() {
        if (closed) {
            return;  // 防止重复关闭
        }
        writeLock.lock();
        try {
            if (closed) {
                return;
            }
            // 强制刷盘
            if (mappedByteBuffer != null) {
                mappedByteBuffer.force();
            }
            // 【核心修复】使用 BufferCleaner 清理内存映射
            if (mappedByteBuffer != null) {
                // 替换不存在的 cleanAndWait 为 clean
                BufferCleaner.clean(mappedByteBuffer);
                mappedByteBuffer = null;

                // 【核心修复】增加强制GC和短暂等待，以确保文件句柄被释放
                // 这对于在测试中立即删除或重新打开文件至关重要
                System.gc();
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
            // 使用资源管理器安全关闭文件通道
            ResourceManager.safeCloseChannel(fileChannel, "MappedFile-" + fileName);
            fileChannel = null;
            // 使用资源管理器安全关闭随机访问文件
            ResourceManager.safeClose(randomAccessFile, "MappedFile-" + fileName);
            randomAccessFile = null;
            closed = true;
            log.info("Closed mapped file: {}", file.getAbsolutePath());
        } catch (Exception e) {
            log.error("Error closing mapped file: {}", file.getAbsolutePath(), e);
        } finally {
            writeLock.unlock();
        }
    }

    /**
     * 删除文件
     */
    public void delete() {
        writeLock.lock();
        try {
            // 先关闭所有资源
            if (!closed) {
                close();
            }
            // 尝试删除文件
            File file = new File(fileName);
            if (file.exists()) {
                if (ResourceManager.safeDeleteFile(file)) {
                    log.info("Deleted file: {}", file.getAbsolutePath());
                } else {
                    log.warn("Failed to delete file: {}", file.getAbsolutePath());
                }
            }
        } catch (Exception e) {
            log.error("Error deleting file: {}", new File(fileName).getAbsolutePath(), e);
        } finally {
            writeLock.unlock();
        }
    }

    /**
     * 获取文件大小
     *
     * @return 文件大小
     */
    public long getFileSize() {
        return mappedFileSize;
    }

    /**
     * 获取写入位置
     *
     * @return 写入位置
     */
    public long getWrotePosition() {
        return wrotePosition.get();
    }

    /**
     * 获取提交位置
     *
     * @return 提交位置
     */
    public long getCommittedPosition() {
        return committedPosition.get();
    }

    /**
     * 获取刷盘位置
     *
     * @return 刷盘位置
     */
    public long getFlushedPosition() {
        return flushedPosition.get();
    }

    /**
     * 回滚写入位置
     *
     * @param length 回滚长度
     */
    public void rollback(long length) {
        synchronized (writeLock) {
            try {
                long currentPosition = wrotePosition.get();
                if (currentPosition >= length) {
                    wrotePosition.addAndGet(-length);
                } else {
                    log.error("Cannot rollback {} bytes in file {}, current position is only {}",
                            length, fileName, currentPosition);
                }
            } catch (Exception e) {
                log.error("Failed to rollback write position in file: {}", fileName, e);
            }
        }
    }

    /**
     * 检查是否有足够的空间写入消息
     *
     * @param totalLength 消息总长度（包括4字节长度前缀）
     * @return 是否有足够空间
     */
    public boolean hasSpace(int totalLength) {
        return wrotePosition.get() + totalLength <= mappedFileSize - 1024;
    }

    /**
     * 获取文件名
     *
     * @return 文件名
     */
    public String getFileName() {
        return file.getAbsolutePath();
    }

    /**
     * 获取当前文件的索引
     *
     * @return 文件索引
     */
    public int getFileIndex() {
        // 从文件名中提取索引，例如 "00000000000000000000" -> 0
        try {
            String name = file.getName();
            String indexStr = name.substring(0, name.lastIndexOf('.'));
            return Integer.parseInt(indexStr);
        } catch (Exception e) {
            log.warn("Failed to parse file index from fileName: {}", fileName);
            return 0;
        }
    }

    /**
     * 获取当前逻辑偏移量
     *
     * @return 当前逻辑偏移量
     */
    public long getCurrentLogicalOffset() {
        return (long) getFileIndex() * mappedFileSize + wrotePosition.get();
    }

    /**
     * 保存写入位置到文件
     */
    public void saveWritePosition() {
        writeLock.lock();
        try {
            // 将写入位置保存到文件末尾
            mappedByteBuffer.position((int) (mappedFileSize - 8));
            mappedByteBuffer.putLong(wrotePosition.get());
            mappedByteBuffer.force();
        } catch (Exception e) {
            log.error("Failed to save write position", e);
        } finally {
            writeLock.unlock();
        }
    }

    /**
     * 从文件恢复写入位置
     */
    public void restoreWritePosition() {
        readLock.lock();
        try {
            // 从文件末尾读取写入位置
            mappedByteBuffer.position((int) (mappedFileSize - 8));
            long savedPosition = mappedByteBuffer.getLong();
            if (savedPosition > 0 && savedPosition < mappedFileSize) {
                wrotePosition.set(savedPosition);
                flushedPosition.set(savedPosition);
            }
        } catch (Exception e) {
            log.error("Failed to restore write position", e);
        } finally {
            readLock.unlock();
        }
    }

    /**
     * 写入消息 - 保持向后兼容性
     *
     * @param message 消息字节数组
     * @return 写入是否成功
     */
    public boolean writeMessage(byte[] message) {
        if (message == null || message.length == 0) {
            log.error("Invalid message: null or empty");
            return false;
        }
        // 计算总长度（消息长度 + 4字节长度前缀）
        int totalLength = message.length + 4;
        // 验证消息长度
        if (totalLength > mappedFileSize - 1024) {
            log.error("Message too large: total length={}, max={}", totalLength, mappedFileSize - 1024);
            return false;
        }
        // 检查是否有足够空间
        long currentPosition = wrotePosition.get();
        if (currentPosition + totalLength > mappedFileSize - 1024) {
            log.error("Not enough space in mapped file: position={}, length={}, max={}",
                    currentPosition, totalLength, mappedFileSize - 1024);
            return false;
        }
        try {
            // 写入长度前缀
            mappedByteBuffer.position((int) currentPosition);
            mappedByteBuffer.putInt(message.length);
            wrotePosition.addAndGet(4);
            // 写入消息内容
            mappedByteBuffer.put(message);
            wrotePosition.addAndGet(message.length);
            // 立即提交
            commit();
            return true;
        } catch (Exception e) {
            log.error("Error writing message, length={}, position={}", message.length, currentPosition, e);
            return false;
        }
    }

    /**
     * 直接写入消息（不添加长度前缀）
     *
     * @param message 消息字节数组（已包含长度前缀）
     * @return 写入是否成功
     */
    public boolean writeMessageDirect(byte[] message) {
        if (message == null || message.length == 0) {
            log.error("Invalid message: null or empty");
            return false;
        }
        // 验证消息长度
        if (message.length > mappedFileSize - 1024) {
            log.error("Message too large: length={}, max={}", message.length, mappedFileSize - 1024);
            return false;
        }
        // 检查是否有足够空间
        long currentPosition = wrotePosition.get();
        if (currentPosition + message.length > mappedFileSize - 1024) {
            log.error("Not enough space in mapped file: position={}, length={}, max={}",
                    currentPosition, message.length, mappedFileSize - 1024);
            return false;
        }
        try {
            // 直接写入消息内容（不添加长度前缀）
            mappedByteBuffer.position((int) currentPosition);
            mappedByteBuffer.put(message);
            wrotePosition.addAndGet(message.length);
            // 立即提交
            commit();
            return true;
        } catch (Exception e) {
            log.error("Error writing message directly, length={}, position={}", message.length, currentPosition, e);
            return false;
        }
    }

}